package com.lxl.testHd.relation.one;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * @author ：e_lixilin
 * @date ：2022/2/23 11:10
 * @description：
 * @modified By： 两个学校的共同专业
 */
public class CommonMajorTwo {
    static class SchoolCommonMajorsMapper extends Mapper<LongWritable, Text, Text, Text> {
        Text school = new Text();
        Text major = new Text();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 西3大学-西6大学	中医学
            String[] fields = value.toString().split("\t");
            school.set(fields[0]);
            major.set(fields[1]);
            context.write(school,major);
        }

    }

    static class SchoolCommonMajorsReducer extends Reducer<Text, Text, Text, Text> {
        Text outVal = new Text();
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            List<String> list = new ArrayList<>();
            for(Text major : values){
                list.add(major.toString()) ;
            }
            outVal.set(list.toString());
            context.write(key,outVal);
        }
    }

    static class SchoolCommonMajorsMapperOne extends Mapper<LongWritable, Text, Text, Text> {
        Text school = new Text();
        Text major = new Text();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 西1大学:软件	戏剧	英语	土木	天文学	数学	考古
            String[] fields = value.toString().split(":");
            school.set(fields[0]);
            String[] majors = fields[1].split("\t");
            for (String str : majors) {
                major.set(str);
                context.write(major,school);
            }
        }

    }

    static class SchoolCommonMajorsReducerOne extends Reducer<Text, Text, Text, Text> {
        Text outKey = new Text();
        Text outVal = new Text();
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            List<String> list = new ArrayList<>();
            for(Text major : values){
                list.add(major.toString()) ;
            }
            //排序
            Collections.sort(list);
            //把values两两组合，并以的“A-B”形式显示
            for(int i=0;i< list.size()-1;i++)
            {
                for(int j=i+1;j< list.size();j++)
                {
                    outKey.set(list.get(i)+"-"+list.get(j) );
                    outVal.set(key);
                    context.write( outKey,outVal);
                }
            }
        }
    }

    public static void main(String[] args) throws Exception{
        args = new String[]{"D:\\big-data\\relation", "D:\\big-data\\relation\\output","D:\\big-data\\relation\\outputTwo"};
        Configuration config=new Configuration();
        FileSystem fs= FileSystem.get(config);
        JobConf conf = new JobConf(CommonMajorTwo.class);
// 第一个job的配置
        @SuppressWarnings("deprecation")
        Job job1 = new Job(conf, "join1");
        job1.setJarByClass(CommonMajorTwo.class);
        job1.setJobName("weibo1");
        job1.setMapperClass(SchoolCommonMajorsMapperOne.class);
        job1.setReducerClass(SchoolCommonMajorsReducerOne.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(Text.class);
// 加入控制容器
        ControlledJob ctrljob1 = new ControlledJob(conf);
        ctrljob1.setJob(job1);
// job1的输入输出文件路径
        FileInputFormat.addInputPath(job1, new Path(args[0]));
        Path path1=new Path(args[1]);
        if(fs.exists(path1)){
            fs.delete(path1,true);
        }
        FileOutputFormat.setOutputPath(job1, path1);
// 第二个作业的配置
        @SuppressWarnings("deprecation")
        Job job2 = new Job(conf, "Join2");
        job2.setJarByClass(CommonMajorTwo.class);
        job2.setJobName("weibo2");
        job2.setMapperClass(SchoolCommonMajorsMapper.class);
//        job2.setCombinerClass(TwoReduce.class);
        job2.setReducerClass(SchoolCommonMajorsReducer.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(Text.class);
// 作业2加入控制容器
        ControlledJob ctrljob2 = new ControlledJob(conf);
        ctrljob2.setJob(job2);
// 设置多个作业直接的依赖关系
// 如下所写：
// 意思为job2的启动，依赖于job1作业的完成
        ctrljob2.addDependingJob(ctrljob1);
// 输入路径是上一个作业的输出路径，因此这里填path1,要和上面对应好
        FileInputFormat.addInputPath(job2, path1);
// 输出路径从新传入一个参数，这里需要注意，因为我们最后的输出文件一定要是没有出现过得
// 因此我们在这里new Path(args[2])因为args[2]在上面没有用过，只要和上面不同就可以了
        Path path2=new Path(args[2]);
        if(fs.exists(path2)){
            fs.delete(path2,true);
        }
        FileOutputFormat.setOutputPath(job2, path2);

        // 主的控制容器，控制上面的总的2个子作业
        JobControl jobCtrl = new JobControl("myctrl");
        // 添加到总的JobControl里，进行控制
        jobCtrl.addJob(ctrljob1);
        jobCtrl.addJob(ctrljob2);
        // 在线程启动，记住一定要有这个
        Thread t = new Thread(jobCtrl);
        t.start();
        while (true) {
            if (jobCtrl.allFinished()) {
                // 如果作业成功完成，就打印成功作业的信息
                System.out.println(jobCtrl.getSuccessfulJobList());
                System.out.println("所有job执行完毕");
                jobCtrl.stop();
                break;
            }
        }
    }
}
